package com.suxin.kafka.config;

import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.boot.autoconfigure.kafka.KafkaProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;

/**
 * @Classname KafKaConfig
 * @Description [ KafKaConfig ]
 * @Author Tang
 * @Date 2020/11/14 16:19
 * @Created by ASUS
 */
@EnableKafka
@Configuration
public class KafKaTwoConfig {

    @Bean
    ConcurrentKafkaListenerContainerFactory<String, String> kafkaListenerTwoContainerFactory(
            @Qualifier("producerConfigs") KafkaProperties consumerTwoConfigs) {
        ConcurrentKafkaListenerContainerFactory<String, String> factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerTwoFactory(consumerTwoConfigs));
        return factory;
    }

    @Bean
    public ConsumerFactory<String, String> consumerTwoFactory(
            @Qualifier("producerConfigs") KafkaProperties consumerTwoConfigs) {
        return new DefaultKafkaConsumerFactory<>(
                consumerTwoConfigs.buildConsumerProperties()
        );
    }

    @ConfigurationProperties(prefix = "sys.kafka.two")
    @Bean
    public KafkaProperties consumerTwoConfigs() {
        return new KafkaProperties();
    }

}